from typing import Optional, Any, Mapping, Iterator, AsyncIterator, Union, Dict

try:
    from enum import StrEnum
except ImportError:
    from strenum import StrEnum
try:
    from http import HTTPMethod
except ImportError:
    class HTTPMethod(StrEnum):
        GET = "GET"
        POST = "POST"
        PUT = "PUT"
        DELETE = "DELETE"

import httpx
# noinspection PyProtectedMember
import httpx._types as types
from httpx_sse import connect_sse, ServerSentEvent, aconnect_sse
from pydantic import BaseModel

from client import errors, models

_httpx_client = httpx.Client()
_async_httpx_client = httpx.AsyncClient()

IGNORED_STREAM_EVENTS = (models.StreamEvent.PING.value,)

# feedback
ENDPOINT_FEEDBACKS = "/messages/{message_id}/feedbacks"
# suggest
ENDPOINT_SUGGESTED = "/messages/{message_id}/suggested"
# files upload
ENDPOINT_FILES_UPLOAD = "/files/upload"
# completion
ENDPOINT_COMPLETION_MESSAGES = "/completion-messages"
ENDPOINT_STOP_COMPLETION_MESSAGES = "/completion-messages/{task_id}/stop"
# chat
ENDPOINT_CHAT_MESSAGES = "/chat-messages"
ENDPOINT_STOP_CHAT_MESSAGES = "/chat-messages/{task_id}/stop"
# workflow
ENDPOINT_RUN_WORKFLOWS = "/workflows/run"
ENDPOINT_STOP_WORKFLOWS = "/workflows/tasks/{task_id}/stop"
# audio <-> text
ENDPOINT_TEXT_TO_AUDIO = "/text-to-audio"
ENDPOINT_AUDIO_TO_TEXT = "/audio-to-text"


class Client(BaseModel):
    api_key: str
    api_base: Optional[str] = "https://api.dify.ai/v1"

    def request(self, endpoint: str, method: str,
                content: Optional[types.RequestContent] = None,
                data: Optional[types.RequestData] = None,
                files: Optional[types.RequestFiles] = None,
                json: Optional[Any] = None,
                params: Optional[types.QueryParamTypes] = None,
                headers: Optional[Mapping[str, str]] = None,
                **kwargs: object,
                ) -> httpx.Response:
        """
        向指定端点发送同步HTTP请求。

        Args:
            endpoint: 要发送请求的API端点。
            method: 要使用的HTTP方法（例如，'GET'，'POST'）。
            content: 要包含在请求体中的原始内容。
            data: 要包含在请求体中的表单数据。
            files: 要包含在请求体中的文件。
            json: 要包含在请求体中的JSON数据。
            params: 要包含在请求URL中的查询参数。
            headers: 要包含在请求中的附加头信息。
            **kwargs: 要传递给请求函数的额外关键字参数。

        Returns:
            包含HTTP响应的`httpx.Response`对象。

        Raises:
            如果响应包含错误，则会引发各种DifyAPIError异常。
        """
        merged_headers = {}
        if headers:
            merged_headers.update(headers)
        self._prepare_auth_headers(merged_headers)

        response = _httpx_client.request(method, endpoint, content=content, data=data, files=files, json=json,
                                         params=params, headers=merged_headers, **kwargs)
        errors.raise_for_status(response)
        return response

    def request_stream(self, endpoint: str, method: str,
                       content: Optional[types.RequestContent] = None,
                       data: Optional[types.RequestData] = None,
                       files: Optional[types.RequestFiles] = None,
                       json: Optional[Any] = None,
                       params: Optional[types.QueryParamTypes] = None,
                       headers: Optional[Mapping[str, str]] = None,
                       **kwargs,
                       ) -> Iterator[ServerSentEvent]:
        """
        打开到指定端点的服务器发送事件（SSE）流。

        Args:
            endpoint: 要发送请求的API端点。
            method: 要使用的HTTP方法（例如，'GET'，'POST'）。
            content: 要包含在请求体中的原始内容。
            data: 要包含在请求体中的表单数据。
            files: 要包含在请求体中的文件。
            json: 要包含在请求体中的JSON数据。
            params: 要包含在请求URL中的查询参数。
            headers: 要包含在请求中的附加头信息。
            **kwargs: 要传递给请求函数的额外关键字参数。

        Returns:
            表示事件流的`ServerSentEvent`对象的迭代器。

        Raises:
            如果在流中收到错误事件，则会引发各种DifyAPIError异常。
        """
        merged_headers = {}
        if headers:
            merged_headers.update(headers)
        self._prepare_auth_headers(merged_headers)

        with connect_sse(_httpx_client, method, endpoint, headers=merged_headers,
                         content=content, data=data, files=files, json=json, params=params, **kwargs) as event_source:
            if not _check_stream_content_type(event_source.response):
                event_source.response.read()
                errors.raise_for_status(event_source.response)
            for sse in event_source.iter_sse():
                errors.raise_for_status(sse)
                if sse.event in IGNORED_STREAM_EVENTS or sse.data in IGNORED_STREAM_EVENTS:
                    continue
                yield sse

    def feedback_messages(self, message_id: str, req: models.FeedbackRequest, **kwargs) -> models.FeedbackResponse:
        """
        为特定消息提交反馈。

        Args:
            message_id: 要提交反馈的消息的标识符。
            req: 包含反馈详情（如评分）的`FeedbackRequest`对象。
            **kwargs: 要传递给请求函数的额外关键字参数。

        Returns:
            包含反馈提交结果的`FeedbackResponse`对象。
        """
        response = self.request(
            self._prepare_url(ENDPOINT_FEEDBACKS, message_id=message_id),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.FeedbackResponse(**response.json())

    def suggest_messages(self, message_id: str, req: models.ChatSuggestRequest, **kwargs) -> models.ChatSuggestResponse:
        """
        根据特定消息检索建议消息。

        Args:
            message_id: 要获取建议的消息的标识符。
            req: 包含请求详情的`ChatSuggestRequest`对象。
            **kwargs: 要传递给请求函数的额外关键字参数。

        Returns:
            包含建议消息的`ChatSuggestResponse`对象。
        """
        response = self.request(
            self._prepare_url(ENDPOINT_SUGGESTED, message_id=message_id),
            HTTPMethod.GET,
            params=req.model_dump(),
            **kwargs,
        )
        return models.ChatSuggestResponse(**response.json())

    def upload_files(self, file: types.FileTypes, req: models.UploadFileRequest,
                     **kwargs) -> models.UploadFileResponse:
        """
        上传文件以供后续请求使用。

        Args:
            file: 要上传的文件。可以是类文件对象，或元组（`文件名`，类文件对象，mime类型）。
            req: 包含上传详情（如上传用户）的`UploadFileRequest`对象。
            **kwargs: 要传递给请求函数的额外关键字参数。

        Returns:
            包含上传文件详情（如标识符和URL）的`UploadFileResponse`对象。
       """
        response = self.request(
            self._prepare_url(ENDPOINT_FILES_UPLOAD),
            HTTPMethod.POST,
            data=req.model_dump(),
            files=[("file", file)],
            **kwargs,
        )
        return models.UploadFileResponse(**response.json())

    def completion_messages(self, req: models.CompletionRequest, **kwargs) \
            -> Union[models.CompletionResponse, Iterator[models.CompletionStreamResponse]]:
        """
        发送请求以基于提供的输入生成一个或一系列补全。

        Returns:
            如果响应模式是阻塞(blocking)的，返回包含生成消息的`CompletionResponse`对象。
            如果响应模式是流式(streaming)的，返回包含生成事件流的`CompletionStreamResponse`对象的迭代器。
       """
        if req.response_mode == models.ResponseMode.BLOCKING:
            return self._completion_messages(req, **kwargs)
        if req.response_mode == models.ResponseMode.STREAMING:
            return self._completion_messages_stream(req, **kwargs)
        raise ValueError(f"Invalid request_mode: {req.response_mode}")

    def _completion_messages(self, req: models.CompletionRequest, **kwargs) -> models.CompletionResponse:
        response = self.request(
            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.CompletionResponse(**response.json())

    def _completion_messages_stream(self, req: models.CompletionRequest, **kwargs) \
            -> Iterator[models.CompletionStreamResponse]:
        event_source = self.request_stream(
            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        for sse in event_source:
            yield models.build_completion_stream_response(sse.json())

    def stop_completion_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
        """
        发送请求以停止流式补全任务。

        Returns:
            指示操作成功的`StopResponse`对象。
       """
        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_COMPLETION_MESSAGES, task_id=task_id), req, **kwargs)

    def chat_messages(self, req: models.ChatRequest, **kwargs) \
            -> Union[models.ChatResponse, Iterator[models.ChatStreamResponse]]:
        """
        发送请求以基于提供的输入生成一个或一系列聊天消息。

        Returns:
            如果响应模式是阻塞的，返回包含生成聊天消息的`ChatResponse`对象。
            如果响应模式是流式的，返回包含聊天事件流的`ChatStreamResponse`对象的迭代器。
        """
        if req.response_mode == models.ResponseMode.BLOCKING:
            return self._chat_messages(req, **kwargs)
        if req.response_mode == models.ResponseMode.STREAMING:
            return self._chat_messages_stream(req, **kwargs)
        raise ValueError(f"Invalid request_mode: {req.response_mode}")

    def _chat_messages(self, req: models.ChatRequest, **kwargs) -> models.ChatResponse:
        response = self.request(
            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.ChatResponse(**response.json())

    def _chat_messages_stream(self, req: models.ChatRequest, **kwargs) -> Iterator[models.ChatStreamResponse]:
        event_source = self.request_stream(
            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        for sse in event_source:
            yield models.build_chat_stream_response(sse.json())

    def stop_chat_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
        """
        发送请求以停止流式聊天任务。

        Returns:
            指示操作成功的`StopResponse`对象。
        """
        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_CHAT_MESSAGES, task_id=task_id), req, **kwargs)

    def run_workflows(self, req: models.WorkflowsRunRequest, **kwargs) \
            -> Union[models.WorkflowsRunResponse, Iterator[models.WorkflowsRunStreamResponse]]:
        """
        启动工作流的执行，工作流可以包含多个步骤和操作。

        Returns:
            如果响应模式是阻塞的，返回包含已完成工作流结果的`WorkflowsRunResponse`对象。
            如果响应模式是流式的，返回包含工作流事件流的`WorkflowsRunStreamResponse`对象的迭代器。
        """
        if req.response_mode == models.ResponseMode.BLOCKING:
            return self._run_workflows(req, **kwargs)
        if req.response_mode == models.ResponseMode.STREAMING:
            return self._run_workflows_stream(req, **kwargs)
        raise ValueError(f"Invalid request_mode: {req.response_mode}")

    def _run_workflows(self, req: models.WorkflowsRunRequest, **kwargs) -> models.WorkflowsRunResponse:
        response = self.request(
            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.WorkflowsRunResponse(**response.json())

    def _run_workflows_stream(self, req: models.WorkflowsRunRequest, **kwargs) \
            -> Iterator[models.WorkflowsRunStreamResponse]:
        event_source = self.request_stream(
            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        for sse in event_source:
            yield models.build_workflows_stream_response(sse.json())

    def stop_workflows(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
        """
        发送请求以停止流式工作流任务。

        Returns:
           指示操作成功的`StopResponse`对象。
        """
        return self._stop_stream(self._prepare_url(ENDPOINT_STOP_WORKFLOWS, task_id=task_id), req, **kwargs)

    def _stop_stream(self, endpoint: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
        response = self.request(
            endpoint,
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.StopResponse(**response.json())

    def _prepare_url(self, endpoint: str, **kwargs) -> str:
        return self.api_base + endpoint.format(**kwargs)

    def _prepare_auth_headers(self, headers: Dict[str, str]):
        if "authorization" not in (key.lower() for key in headers.keys()):
            headers["Authorization"] = f"Bearer {self.api_key}"


# 异步客户端
class AsyncClient(BaseModel):
    api_key: str
    api_base: Optional[str] = "https://api.dify.ai/v1"

    async def arequest(self, endpoint: str, method: str,
                       content: Optional[types.RequestContent] = None,
                       data: Optional[types.RequestData] = None,
                       files: Optional[types.RequestFiles] = None,
                       json: Optional[Any] = None,
                       params: Optional[types.QueryParamTypes] = None,
                       headers: Optional[Mapping[str, str]] = None,
                       **kwargs,
                       ) -> httpx.Response:
        """
        异步向指定的Dify API端点发送请求。

        Args:
            endpoint: 发送请求的端点URL。
            method: 用于请求的HTTP方法（例如，'GET'，'POST'）。
            content: 包含在请求体中的原始内容（如有）。
            data: 要在请求体中发送的表单数据。
            files: 要随请求上传的文件。
            json: 要在请求体中发送的JSON数据。
            params: 要包含在请求URL中的查询参数。
            headers: 要随请求发送的附加头信息。
            **kwargs: 要传递给底层HTTPX请求函数的额外关键字参数。

        Returns:
            包含服务器对HTTP请求的响应的httpx.Response对象。

        Raises:
            如果响应包含错误，则会引发各种DifyAPIError异常。
       """
        merged_headers = {}
        if headers:
            merged_headers.update(headers)
        self._prepare_auth_headers(merged_headers)

        response = await _async_httpx_client.request(method, endpoint, content=content, data=data, files=files,
                                                     json=json, params=params, headers=merged_headers, **kwargs)
        errors.raise_for_status(response)
        return response

    async def arequest_stream(self, endpoint: str, method: str,
                              content: Optional[types.RequestContent] = None,
                              data: Optional[types.RequestData] = None,
                              files: Optional[types.RequestFiles] = None,
                              json: Optional[Any] = None,
                              params: Optional[types.QueryParamTypes] = None,
                              headers: Optional[Mapping[str, str]] = None,
                              **kwargs,
                              ) -> AsyncIterator[ServerSentEvent]:
        """
         异步建立到指定Dify API端点的流式连接。

        Args:
           endpoint: 发送请求的端点URL。
            method: 用于请求的HTTP方法（例如，'GET'，'POST'）。
            content: 包含在请求体中的原始内容（如有）。
            data: 要在请求体中发送的表单数据。
            files: 要随请求上传的文件。
            json: 要在请求体中发送的JSON数据。
            params: 要包含在请求URL中的查询参数。
            headers: 要随请求发送的附加头信息。
            **kwargs: 要传递给底层HTTPX请求函数的额外关键字参数。

        Yields:
             表示从服务器接收的事件的ServerSentEvent对象。

        Raises:
           如果在流中收到错误事件，则会引发各种DifyAPIError异常。
       """
        merged_headers = {}
        if headers:
            merged_headers.update(headers)
        self._prepare_auth_headers(merged_headers)

        async with aconnect_sse(_async_httpx_client, method, endpoint, headers=merged_headers,
                                content=content, data=data, files=files, json=json, params=params,
                                **kwargs) as event_source:
            if not _check_stream_content_type(event_source.response):
                await event_source.response.aread()
                errors.raise_for_status(event_source.response)
            async for sse in event_source.aiter_sse():
                errors.raise_for_status(sse)
                if sse.event in IGNORED_STREAM_EVENTS or sse.data in IGNORED_STREAM_EVENTS:
                    continue
                yield sse

    async def afeedback_messages(self, message_id: str, req: models.FeedbackRequest, **kwargs) \
            -> models.FeedbackResponse:
        """
       为特定消息提交反馈。

        Args:
            message_id: 要提交反馈的消息的标识符。
            req: 包含反馈详情（如评分）的`FeedbackRequest`对象。
            **kwargs: 要传递给请求函数的额外关键字参数。

        Returns:
            包含反馈提交结果的`FeedbackResponse`对象。
        """
        response = await self.arequest(
            self._prepare_url(ENDPOINT_FEEDBACKS, message_id=message_id),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.FeedbackResponse(**response.json())

    async def asuggest_messages(self, message_id: str, req: models.ChatSuggestRequest, **kwargs) \
            -> models.ChatSuggestResponse:
        """
        根据特定消息检索建议消息。

        Args:
            message_id: 要提交反馈的消息的标识符。
            req: 包含反馈详情（如评分）的`FeedbackRequest`对象。
            **kwargs: 要传递给请求函数的额外关键字参数。

        Returns:
            包含反馈提交结果的`ChatSuggestResponse`对象。
        """
        response = await self.arequest(
            self._prepare_url(ENDPOINT_SUGGESTED, message_id=message_id),
            HTTPMethod.GET,
            params=req.model_dump(),
            **kwargs,
        )
        return models.ChatSuggestResponse(**response.json())

    async def aupload_files(self, file: types.FileTypes, req: models.UploadFileRequest, **kwargs) \
            -> models.UploadFileResponse:
        """
        上传一个文件，以便在后续请求中使用。

        Args:
            file: 要上传的文件。可以是类文件对象，或者是一个元组，格式为
           （`文件名`，类文件对象，mime类型）。
            req: 一个`UploadFileRequest`对象，包含上传的详细信息，例如上传用户等。
            **kwargs: 传递给请求函数的额外关键字参数。

        Returns:
            一个`UploadFileResponse`对象，包含已上传文件的详细信息，例如文件标识符和URL等。
        """
        response = await self.arequest(
            self._prepare_url(ENDPOINT_FILES_UPLOAD),
            HTTPMethod.POST,
            data=req.model_dump(),
            files=[("file", file)],
            **kwargs,
        )
        return models.UploadFileResponse(**response.json())

    async def acompletion_messages(self, req: models.CompletionRequest, **kwargs) \
            -> Union[models.CompletionResponse, AsyncIterator[models.CompletionStreamResponse]]:
        """
        发送请求以基于提供的输入生成一个或一系列补全内容。

        Returns:
            如果响应模式为阻塞模式，返回一个 `CompletionResponse` 对象，包含生成的消息。
            如果响应模式为流式模式，返回一个 `CompletionStreamResponse` 对象的迭代器，包含
            生成事件的流。
        """
        if req.response_mode == models.ResponseMode.BLOCKING:
            return await self._acompletion_messages(req, **kwargs)
        if req.response_mode == models.ResponseMode.STREAMING:
            return self._acompletion_messages_stream(req, **kwargs)
        raise ValueError(f"Invalid request_mode: {req.response_mode}")

    async def _acompletion_messages(self, req: models.CompletionRequest, **kwargs) -> models.CompletionResponse:
        response = await self.arequest(
            self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.CompletionResponse(**response.json())

    async def _acompletion_messages_stream(self, req: models.CompletionRequest, **kwargs) \
            -> AsyncIterator[models.CompletionStreamResponse]:
        async for sse in self.arequest_stream(
                self._prepare_url(ENDPOINT_COMPLETION_MESSAGES),
                HTTPMethod.POST,
                json=req.model_dump(),
                **kwargs):
            yield models.build_completion_stream_response(sse.json())

    async def astop_completion_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
        """
        发送停止流式传输完成任务的请求。

        Returns:
            A `StopResponse` object indicating the success of the operation.
        """
        return await self._astop_stream(
            self._prepare_url(ENDPOINT_STOP_COMPLETION_MESSAGES, task_id=task_id), req, **kwargs)

    async def achat_messages(self, req: models.ChatRequest, **kwargs) \
            -> Union[models.ChatResponse, AsyncIterator[models.ChatStreamResponse]]:
        """
        发送请求以基于提供的输入生成一条或一系列聊天消息。
        Returns:
            如果响应模式为阻塞模式，返回一个 `ChatResponse` 对象，包含生成的聊天消息。
            如果响应模式为流式模式，返回一个 `ChatStreamResponse` 对象的迭代器，包含
            聊天事件的流。
        """
        if req.response_mode == models.ResponseMode.BLOCKING:
            return await self._achat_messages(req, **kwargs)
        if req.response_mode == models.ResponseMode.STREAMING:
            return self._achat_messages_stream(req, **kwargs)
        raise ValueError(f"Invalid request_mode: {req.response_mode}")

    async def _achat_messages(self, req: models.ChatRequest, **kwargs) -> models.ChatResponse:
        response = await self.arequest(
            self._prepare_url(ENDPOINT_CHAT_MESSAGES),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.ChatResponse(**response.json())

    async def _achat_messages_stream(self, req: models.ChatRequest, **kwargs) \
            -> AsyncIterator[models.ChatStreamResponse]:
        async for sse in self.arequest_stream(
                self._prepare_url(ENDPOINT_CHAT_MESSAGES),
                HTTPMethod.POST,
                json=req.model_dump(),
                **kwargs):
            yield models.build_chat_stream_response(sse.json())

    async def astop_chat_messages(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
        """
        发送请求以停止流式聊天任务。

        Returns:
           一个 `StopResponse` 对象，指示操作是否成功。
       """
        return await self._astop_stream(self._prepare_url(ENDPOINT_STOP_CHAT_MESSAGES, task_id=task_id), req, **kwargs)

    async def arun_workflows(self, req: models.WorkflowsRunRequest, **kwargs) \
            -> Union[models.WorkflowsRunResponse, AsyncIterator[models.WorkflowsStreamResponse]]:
        """
        异步启动一个工作流的执行，该工作流可包含多个步骤和操作。

        Returns:
            如果响应模式为阻塞模式，返回一个 `WorkflowsRunResponse` 对象，包含已完成工作流的结果。
            如果响应模式为流式模式，返回一个 `WorkflowsRunStreamResponse` 对象的异步迭代器，
            包含工作流事件的流。
        """
        if req.response_mode == models.ResponseMode.BLOCKING:
            return await self._arun_workflows(req, **kwargs)
        if req.response_mode == models.ResponseMode.STREAMING:
            return self._arun_workflows_stream(req, **kwargs)
        raise ValueError(f"Invalid request_mode: {req.response_mode}")

    async def _arun_workflows(self, req: models.WorkflowsRunRequest, **kwargs) -> models.WorkflowsRunResponse:
        response = await self.arequest(
            self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.WorkflowsRunResponse(**response.json())

    async def _arun_workflows_stream(self, req: models.WorkflowsRunRequest, **kwargs) \
            -> AsyncIterator[models.WorkflowsRunStreamResponse]:
        async for sse in self.arequest_stream(
                self._prepare_url(ENDPOINT_RUN_WORKFLOWS),
                HTTPMethod.POST,
                json=req.model_dump(),
                **kwargs):
            yield models.build_workflows_stream_response(sse.json())

    async def astop_workflows(self, task_id: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
        """
        发送停止流式工作流任务的请求。

        Returns:
          一个 `StopResponse` 对象，指示操作是否成功。
        """
        return await self._astop_stream(self._prepare_url(ENDPOINT_STOP_WORKFLOWS, task_id=task_id), req, **kwargs)

    async def _astop_stream(self, endpoint: str, req: models.StopRequest, **kwargs) -> models.StopResponse:
        response = await self.arequest(
            endpoint,
            HTTPMethod.POST,
            json=req.model_dump(),
            **kwargs,
        )
        return models.StopResponse(**response.json())

    def _prepare_url(self, endpoint: str, **kwargs) -> str:
        return self.api_base + endpoint.format(**kwargs)

    def _prepare_auth_headers(self, headers: Dict[str, str]):
        if "authorization" not in (key.lower() for key in headers.keys()):
            headers["Authorization"] = f"Bearer {self.api_key}"


def _get_content_type(headers: httpx.Headers) -> str:
    return headers.get("content-type", "").partition(";")[0]


def _check_stream_content_type(response: httpx.Response) -> bool:
    content_type = _get_content_type(response.headers)
    return response.is_success and "text/event-stream" in content_type
